package com.yunji.framework_template.web.config;

import java.util.List;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import lombok.extern.slf4j.Slf4j;

@Slf4j
//@Configuration
public class RocketMQConfig {
	@Value("#{settings['rocketmq.namesevers']}")
	private String namesevers;
	@Value("#{settings['rocketmq.groupName']}")
	private String groupName;
	@Value("#{settings['rocketmq.requestTopic']}")
	private String requestTopic;
	@Value("#{settings['rocketmq.reponseTopic']}")
	private String reponseTopic;

	@Bean(destroyMethod = "shutdown", initMethod = "start")
	public DefaultMQProducer rocketMQProducer() {
		DefaultMQProducer mqProducer = new DefaultMQProducer(groupName);
		try {
			mqProducer.setNamesrvAddr(namesevers);
		} catch (Exception e) {
			log.error("初使化MQProducer发生异常，不会影响应用的启动，如果应用中不涉及到MQ操作可以忽略该异常：" + e.getMessage(), e);
		}
		return mqProducer;
	}

	@Bean(destroyMethod = "shutdown", initMethod = "start")
	public DefaultMQPushConsumer rocketMQConsumer() throws MQClientException {
		DefaultMQPushConsumer mqConsumer = new DefaultMQPushConsumer(groupName);
		try {
			mqConsumer.setNamesrvAddr(namesevers);
			mqConsumer.setMessageListener(new MessageListenerConcurrently() {

				@Override
				public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
						ConsumeConcurrentlyContext context) {
					System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
					return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
				}

			});
			/*
			 * Specify where to start in case the specified consumer group is a brand new
			 * one.
			 */
			mqConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
			/*
			 * Subscribe one more more topics to consume.
			 */
			mqConsumer.subscribe(reponseTopic, "*");
		} catch (Exception e) {
			log.error("初使化MQConsumer发生异常，不会影响应用的启动，如果应用中不涉及到MQ操作可以忽略该异常：" + e.getMessage(), e);
		}
		return mqConsumer;
	}

}
